Outbox 事件路由

直接发送消息到 Kafka 有什么问题

一个典型的业务动作往往是“更新数据库，再发一条消息通知下游”：订单支付成功，更新订单状态的同时往消息队列发一条“订单已支付”。这两个动作分属数据库与消息中间件，没法放进同一个事务，于是无论先做哪个，都有失败窗口：

• 先发消息、后写库：写库失败（约束冲突、连接中断、进程崩溃），消息已经出去，下游收到一条并不存在的“幽灵事件”。
• 先写库、后发消息：发消息失败（网络抖动、broker 不可用），这条业务变更永远不会被下游感知，数据悄悄不一致。

这就是双写问题：两个系统各写一半，没有机制保证“同时成功或同时失败”。用分布式事务强行包裹两边，性能与可用性的代价在绝大多数场景不可接受。

Outbox 模式：先写库，再投递

解法是把“发消息”拆成两步：业务本地事务里只做“更新业务表、往发件箱表插一条待发记录”，两步同事务，要么都成功要么都回滚；事务提交后，再由独立的投递器把发件箱里的记录发布到消息队列：

BEGIN;
  UPDATE 订单表 SET 支付状态 = 已支付 WHERE 订单号 = ...;
  INSERT INTO outbox (聚合类型, 聚合标识, 事件类型, 事件内容) VALUES (...);
COMMIT;

上图就是完整链路：生产方在本地事务中写业务表与发件箱（左虚线框），投递器把待发消息发布到消息队列，消费端先落收件箱去重、再在本地事务中完成业务处理（右虚线框）。两个虚线框是仅有的两处本地事务边界，整个方案不依赖分布式事务。

可靠性由三段共同保证：

环节	目标	机制
生产端	业务成功则消息一定发出	Outbox 模式加本地事务
传输端	消息不丢	MQ 持久化、副本与 ACK
消费端	消息收到则一定处理，且只处理一次	Inbox 模式加幂等消费

本文聚焦生产端的落地：用 Debezium CDC 捕获发件箱表，把每条记录变成干净的 Kafka 事件。消费端 Inbox 是对称的另一半，思路见结构图右半部分，不在本文展开。

发件箱的投递器有两种常见实现：

实现方式	原理	特点
后台轮询	定时扫描待发送状态的记录，发布后标记已发送	实现直接，扫描要配部分索引才不拖垮数据库，延迟秒级
CDC 捕获	监听发件箱表的变更日志，变更即投递	近实时，业务代码零侵入，无轮询压力，本文采用

选定 CDC 投递后，也许会想：为什么不直接捕获业务表，省掉发件箱中转？直接捕获业务表适合“数据同步”类场景（把表镜像到 Elasticsearch、缓存），做“业务事件通知”类场景（订单已支付、订单已取消，要通知下游多个系统）会有几个问题：

• 消息形态脏：每条消息带 before/after/source 包裹，消费端要先剥壳再挑字段。
• 中间态噪音：一次业务操作可能 UPDATE 同一行多次，每次都产生消息，下游收到一堆中间状态；真正有业务含义的“状态跃迁”淹没在里面。
• 契约不稳定：业务表加一列，消息结构就变，所有消费端跟着改；数据库存储结构与消息契约强耦合。

事件从发件箱表走，业务显式声明“什么算一次事件”，这三个问题都从源头消解。

第一步：设计 outbox 表

表结构没有强制标准，核心是五列：事件唯一标识、聚合类型、聚合标识、事件类型、事件内容。EventRouter 按列名取值，列名可自定义，配置里对上即可（本文的列名与第三步配置一一对应）：

CREATE TABLE public.outbox (
    message_id     uuid PRIMARY KEY DEFAULT uuidv7(),
    -- 事件归属的聚合类型，决定消息发往哪个 topic（订单、支付、用户等）
    aggregate_type text        NOT NULL,
    -- 聚合的业务标识，作为消息 key，同一聚合的事件保证顺序
    aggregate_id   text        NOT NULL,
    -- 事件类型（已创建、已支付、已取消等），按需放进消息 header
    event_type     text        NOT NULL,
    -- 事件内容，JSON 文档，消费端直接反序列化的部分
    payload        jsonb       NOT NULL,
    -- 以下为业务自留的发送状态管理列，CDC 不消费，按需增减
    status         text        NOT NULL DEFAULT 'PENDING',
    created_at     timestamptz NOT NULL DEFAULT now(),
    updated_at     timestamptz NOT NULL DEFAULT now(),
    sent_at        timestamptz,
    retry_count    integer     NOT NULL DEFAULT 0,
    last_error     text,
    next_retry_at  timestamptz,
    version        integer     NOT NULL DEFAULT 0,
    remark         text
);

设计要点：

• 主键：事件唯一标识，建议时间有序的 UUIDv7（索引写入友好，避免 UUIDv4 的随机写放大）。uuidv7() 是 PG18 起内置的函数，旧版本可换 gen_random_uuid()（PG13+ 内置）。
• 只插入：outbox 表的约定用法是只 INSERT。EventRouter 也只路由插入事件，后续 UPDATE（比如标记已发送）、DELETE（清理历史行）都不会产生消息，见第四步的验证。
• 业务管理列随便加：status、retry_count、sent_at 这类列是给业务侧的补偿任务用的（扫描未发送、失败重试、定期清理），EventRouter 不会把它们带进消息，不影响事件契约。
• 事件内容用 jsonb：类型本身校验 JSON 合法性，且配合 expand.json.payload 可以在消息里直接展开为 JSON 对象（见配置详解）。

第二步：准备账号与 publication

账号、数据库前置条件与 PostgreSQL 连接器 完全一致，直接沿用。唯一要注意的还是那个 PG15+ 的坑：最小权限账号无权建 publication，必须用超管预建，且名字要与连接器配置里的 publication.name 完全一致：

-- 用超管账号（本环境为 admin）在目标库执行
CREATE PUBLICATION order_db_outbox FOR TABLE public.outbox;

publication 是服务端第一层过滤（决定 WAL 推送哪些表的变更），连接器的表白名单是第二层（客户端过滤）。outbox 模式通常只捕获这一张表，推荐上面这种显式表清单的写法，范围最小；若同一连接器将来还要捕获别的表，用 ALTER PUBLICATION ... ADD TABLE 维护。FOR ALL TABLES 模式也能用（新表自动纳入，但库内所有表的变更都会走一遍服务端过滤，而且无法 ADD/DROP TABLE），按需取舍。

名字必须完全一致

Debezium 启动时按 publication.name 的值查找，找到就复用，找不到就尝试自动创建（然后因权限不足失败）。预建的名字和配置对不上时，预建等于白建。

第三步：注册连接器

还差最后一块拼图：Debezium 的默认输出是整行镜像，发件箱表的一行会原样投出（照旧带 before/after 包裹）。EventRouter 是 Connect 端的单消息变换，在投递前把发件箱行转换成干净的事件消息：

• 消息 key = 聚合标识（同一聚合的事件落同一分区，保序）
• 消息 value = 事件内容本身，不再有包裹结构
• 消息 header = 事件唯一标识、事件类型等元信息
• topic 名 = 按聚合类型动态路由

消费端按聚合类型选 topic，直接反序列化事件内容即可。这个转换在连接器里通过一组 transforms.outbox.* 配置完成，先把完整 JSON 存成 register-outbox.json：

{
  "name": "pg-outbox-connector.order_db.public",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",

    "tasks.max": "1",

    "database.hostname": "host.docker.internal",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "debezium@123",
    "database.dbname": "order_db",

    "plugin.name": "pgoutput",

    "publication.name": "order_db_outbox",
    "publication.autocreate.mode": "disabled",

    "slot.name": "public_outbox",

    "table.include.list": "public.outbox",

    "topic.prefix": "order_db",
    "topic.creation.default.replication.factor": "1",
    "topic.creation.default.partitions": "1",

    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",

    "transforms.outbox.table.field.event.id": "message_id",
    "transforms.outbox.table.field.event.key": "aggregate_id",
    "transforms.outbox.table.field.event.payload": "payload",

    "transforms.outbox.table.fields.additional.placement": "event_type:header:eventType",
    "transforms.outbox.table.expand.json.payload": true,

    "transforms.outbox.route.by.field": "aggregate_type",
    "transforms.outbox.route.topic.replacement": "${routedByValue}"
  }
}

注册与状态检查：

$ curl -i -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d @register-outbox.json

HTTP/1.1 201 Created

$ curl -s http://localhost:8083/connectors/pg-outbox-connector.order_db.public/status

{
  "name": "pg-outbox-connector.order_db.public",
  "connector": {
    "state": "RUNNING",
    "worker_id": "172.20.0.3:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "172.20.0.3:8083"
    }
  ],
  "type": "source"
}

connector.state 和 tasks[].state 都是 RUNNING 才算成功，FAILED 时 tasks[].trace 里是报错堆栈，对照文末 踩坑总结。

数据库连接、plugin.name、slot.name、publication.name、decimal.handling.mode、topic.creation.* 这些通用配置与 PostgreSQL 连接器 相同，下面只讲 EventRouter 特有的部分。

config.topic.prefix

outbox 模式下数据 topic 的命名不再走 ${topic.prefix}.${schema}.${表名} 规则，而是完全由路由规则（后述 route.topic.replacement）决定。

但是 topic.prefix 仍必填，作为连接器的逻辑命名空间使用（位移与元数据隔离），集群内保持唯一即可。

config.transforms

启用单消息变换（SMT）链，outbox 是本连接器里变换器的别名，后面的配置项都挂在这个别名下。别名可自定义，一致即可。

config.transforms.outbox.type

变换器实现类，固定为 io.debezium.transforms.outbox.EventRouter，Debezium 自带，无需额外安装。

config.transforms.outbox.table.field.event.id

事件唯一标识列（即事件ID，消费端可根据该ID判断是否重复消费），值会放进消息 header 的 id。

默认列名是 id，如果不一致时需要明确显示指定（如该示例表主键字段为 message_id）。列名对不上时，任务会在第一条消息处 FAILED，报错文本相当直白：

Caused by: org.apache.kafka.connect.errors.DataException: id is not a valid field name

config.transforms.outbox.table.field.event.key

聚合标识列，值作为消息 key（即消息路由 Key）。

Kafka 按 key 哈希分区，同一聚合（同一订单、同一支付单）的全部事件落同一分区、保持先后顺序，这是下游正确处理事件序列的前提。

默认列名为 aggregateid（无下划线），自定义列名同样要显式指定。

config.transforms.outbox.table.field.event.payload

事件内容列，值作为消息 value。默认列名 payload，与表一致时可省略，但是仍建议显示指定。

config.transforms.outbox.route.by.field

路由依据列，按聚合类型决定消息发往哪个 topic。默认列名为 aggregatetype（无下划线），该示例显示指定字段为 aggregate_type。

config.transforms.outbox.route.topic.replacement

消息 topic 名模板，${routedByValue} 会被路由列（config.transforms.outbox.route.by.field）的值替换。本例直接用聚合类型的值当 topic 名（orders、payments 等），消费端按聚合订阅。

默认模板是 outbox.event.${routedByValue}，想要统一前缀时用默认值即可。

config.transforms.outbox.table.fields.additional.placement

把表中其他列按需附加到消息里，语法是 列名:放置位置:自定义名称，多条逗号分隔。放置位置支持三种：

位置	效果
header	放进消息 header（本例：事件类型列放为 header 的 eventType）
envelope	并入消息 value，与事件内容平级
partition	作为分区计算值（覆盖 key 哈希分区，少用）

同一行事件数据在两种位置下的实际消息效果（三段分别是 header、key、value，之间用制表符分隔）：

header 版（本例配置，事件类型进 header）：

# 消息头信息
id:01a1008a-546f-7a64-9047-b55a1328ef06,eventType:paid	9191642889

# 消息报文信息
{
  "orderId": 9191642889,
  "payableAmount": 199.0
}

envelope 版（事件类型并入 value）：

# 消息头信息
id:01a100a3-166b-7749-beed-654d87a4e01e	9191642889

# 消息报文信息
{
  "payload": {
    "orderId": 9191642889,
    "payableAmount": 199.0
  },
  "eventType": "paid"
}

位置	对比
header	事件类型在 header 里，value 是纯事件内容。消费端不反序列化 value 就能按事件类型分发（路由、过滤、打日志），消息主体保持干净，首选。
envelope	header 里只剩事件唯一标识（id 总是进 header，不受此配置影响），value 的结构变了：事件内容被收进 payload 字段，事件类型与它平级。适合消费端框架不方便读 header 的场景，代价是取数据多一层字段。

config.transforms.outbox.table.expand.json.payload

jsonb 事件内容列的关键配置。不开启时，jsonb 的值在消息里是一段转义后的 JSON 字符串（value 本身是字符串，内容才是 JSON），消费端要二次解析：

"{\"orderId\": 9191642889, \"payableAmount\": 199.0}"

开启后展开为真正的 JSON 对象，消费端直接反序列化：

{"orderId": 9191642889, "payableAmount": 199.0}

别用 JsonData 展开 jsonb

另一组配置 transforms.outbox.format.value 配 JsonData 也能“解析 JSON”，但那是给事件内容列本身是字符串类型（varchar 存 JSON 文本）的场景用的。列是 jsonb 时应当用 expand.json.payload，两者混用会出现转换报错或展开不生效。

transforms.outbox.table.field.event.timestamp

可以指定事件时间列，但要求该列是 BIGINT 毫秒时间戳，timestamptz/timestamp 这类数据库时间类型不行，配了会在第一条消息处 FAILED：

Caused by: org.apache.kafka.connect.errors.DataException: Field 'created_at' is not of type INT64

实际业务并不推荐配置该字段，可直接将业务时间放在消息体中供消费者使用。但如果真需要该字段的话，可以考虑在表中增加一列 UTC 毫秒值。

第四步：数据验证

4.1 插入事件

$ docker exec postgres18 psql -U admin -d order_db -c \
  "INSERT INTO public.outbox (aggregate_type, aggregate_id, event_type, payload) \
   VALUES ('orders', '9191642889', 'paid', '{\"orderId\":9191642889,\"payableAmount\":199.00}');"

topic 名就是聚合类型的值，直接消费（print.headers、print.key 分别把 header 和 key 打出来，与 value 之间用制表符分隔）：

$ docker exec kafka sh -c "/opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:19092 --topic orders \
  --from-beginning --timeout-ms 4000 \
  --property print.headers=true --property print.key=true"

id:01a1008a-546f-7a64-9047-b55a1328ef06,eventType:paid	9191642889	{"orderId":9191642889,"payableAmount":199.0}

一条消息三段结构，对照连接器配置：

段	值	来源
header	id:01a1008a-...、eventType:paid	事件唯一标识列加 additional.placement 附加的事件类型列
key	9191642889	聚合标识列，决定分区与顺序
value	{"orderId":9191642889,"payableAmount":199.0}	事件内容列，已展开为 JSON 对象

199.00 显示为 199.0 是 JSON 数字的规范表示，不是精度丢失。

4.2 更新与删除：不产生事件

outbox 表的约定用法是只插入，但业务侧往往有后台任务要标记发送状态、清理历史行。实测对 outbox 行执行 UPDATE（标记已发送）与 DELETE（清理），topic 均无新消息，任务保持 RUNNING。

EventRouter 只路由插入事件，状态更新与清理动作不会变成重复事件，也不影响链路。业务可以放心地在表上维护发送状态。

4.3 新聚合类型：自动路由新 topic

插入一条聚合类型为 payments 的事件：

$ docker exec postgres18 psql -U admin -d order_db -c \
  "INSERT INTO public.outbox (aggregate_type, aggregate_id, event_type, payload) \
   VALUES ('payments', 'pay-889900', 'PaymentSucceeded', '{\"paymentId\":\"pay-889900\",\"channel\":\"wallet\"}');"

topic 列表里自动出现 payments（由 topic.creation.* 按连接器配置创建）：

$ docker exec kafka sh -c "/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:19092 --list"

debezium_configs
debezium_offsets
debezium_statuses
orders
payments

消费 payments：

id:01a10090-2bef-7d43-b64d-532bf1b81723,eventType:PaymentSucceeded	pay-889900	{"channel":"wallet","paymentId":"pay-889900"}

危险

产线不建议开启自动创建 topic 功能，可考虑先提前创建好 topic 供 outbox 使用。

新增一种聚合类型不需要改任何配置，插入即路由。要注意聚合类型的值会直接成为 topic 名，业务上应约定一个稳定的受控枚举（全小写、复数名词），避免拼出 OrderPaid_v2 这类随意值，topic 会越建越多。

4.4 存量数据

连接器首次注册且快照模式为默认 initial 时，outbox 表里的存量行会以快照事件全部发出（对消费端表现为补发历史事件）。推荐先建表、再注册连接器，让表在注册时是空的，事件从注册时刻开始；如果注册时已有历史数据又不希望补发，把快照模式改为只同步表结构：

"snapshot.mode": "schema_only"

消息结构速查

部件	内容	说明
消息 key	聚合标识列的值	同一聚合的事件落同一分区、分区内保序；消费端按聚合串行处理
header id	事件唯一标识列的值	全局唯一，消费端幂等去重用它
header eventType	事件类型列的值（由 additional.placement 附加）	不解析 value 即可分发；名称可自定义
value	事件内容列的 JSON（expand.json.payload 开启后为对象）	消息主体，只含业务事件数据
topic 名	路由列的值按 route.topic.replacement 模板渲染	按聚合类型分 topic，消费端按需订阅

与直接捕获业务表的消息对比：

维度	直接捕获业务表	outbox 加 EventRouter
消息形态	before/after/source 包裹的整行镜像	纯事件内容
消息数量	每次行变更一条，含大量中间态	业务决定何时算事件，一行一条
表结构变更影响	加列即改消息结构，消费端跟着改	事件内容是显式契约，表结构内部变化不影响
典型用途	数据同步（镜像表到搜索、缓存）	业务事件通知（下游集成、触发流程）

踩坑总结与最佳实践

现象	原因	解法
任务 FAILED：id is not a valid field name	事件唯一标识列配了不存在的列名（默认 id，表里是别的名字）	table.field.event.id 显式指定真实主键列
任务 FAILED：Field 'xxx' is not of type INT64	时间戳列配了数据库时间类型，要求是 BIGINT 毫秒值	删掉 table.field.event.timestamp，或改用毫秒列
value 是一段转义 JSON 字符串，不是对象	jsonb 列默认按字符串传递	配 table.expand.json.payload=true
配了 JsonData 后报转换错误或仍不展开	JsonData 面向字符串类型的事件内容列，jsonb 不适用	jsonb 用 expand.json.payload，别配 format.value
任务 FAILED：permission denied for database 或提示创建 publication	最小权限账号无权自动建 publication，或预建名字与配置不一致	超管预建同名 publication，配 publication.autocreate.mode=disabled
新连接器起不来，复制槽被占用	多个连接器配了同一个 slot.name	一个槽只能被一个连接器独占，换槽名并清理废弃槽
注册后补发了一堆历史事件	表在注册前已有存量数据，初始快照全量发出	先建表后注册；或 snapshot.mode=schema_only
topic 越积越多	路由列的值直接成为 topic 名，值不受控	聚合类型用受控枚举，值全小写复数

推荐的基线连接器配置（单节点集群，jsonb 事件内容，事件类型放 header）：

{
  "name": "pg-outbox-connector.库名.public",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "database.hostname": "host.docker.internal",
    "database.port": "端口(默认5432)",
    "database.user": "账号",
    "database.password": "密码",
    "database.dbname": "库名",
    "plugin.name": "pgoutput",
    "topic.prefix": "库名",
    "publication.name": "预建的 publication 名",
    "publication.autocreate.mode": "disabled",
    "slot.name": "{环境}_{用途}_{唯一标识}",
    "table.include.list": "public.outbox",
    "decimal.handling.mode": "string",
    "topic.creation.default.replication.factor": "1",
    "topic.creation.default.partitions": "1",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.table.field.event.id": "事件唯一标识列",
    "transforms.outbox.table.field.event.key": "聚合标识列",
    "transforms.outbox.table.field.event.payload": "事件内容列",
    "transforms.outbox.route.by.field": "聚合类型列",
    "transforms.outbox.route.topic.replacement": "${routedByValue}",
    "transforms.outbox.table.fields.additional.placement": "事件类型列:header:eventType",
    "transforms.outbox.table.expand.json.payload": true
  }
}

业务侧配套约定：

• 事件内容是契约：字段只增不改语义，消费端按需取字段，做到新增字段不破坏老消费者。
• 同一事务写入：业务表变更与 outbox 插入必须在同一事务，这是模式的一致性根基。
• 幂等消费：至少一次投递语义下，重复不可避免，消费端用 header 里的 id 做去重。
• 及时清理：outbox 表只增不减会持续膨胀，事件确认投递后由后台任务定期清理（清理不产生事件，放心删）。

参考来源

• Debezium: Outbox Event Router：路由器全部配置项、默认值与 jsonb 展开行为说明。
• Transactional Outbox 模式：模式的动机与一般化描述。
• PostgreSQL 连接器：本文的前置篇，账号、publication、通用配置项的细节都在其中。